package cn.iotab.iotkit.transport.api.common.queue;

import cn.iotab.iotkit.queue.queue.QueueClientSetting;
import cn.iotab.iotkit.queue.queue.QueueConsumer;
import cn.iotab.iotkit.queue.queue.QueueProducer;
import cn.iotab.iotkit.queue.queue.provider.kafka.KafkaAdmin;
import cn.iotab.iotkit.queue.queue.provider.kafka.KafkaSetting;
import cn.iotab.iotkit.queue.queue.provider.kafka.factory.KafkaQueueAdminFactory;
import cn.iotab.iotkit.queue.queue.provider.kafka.factory.KafkaQueueFactory;
import cn.iotab.iotkit.queue.rpc.QueueRequestTemplate;
import cn.iotab.iotkit.queue.rpc.QueueResponseTemplate;
import cn.iotab.iotkit.queue.rpc.factory.DefaultQueueRpcFactory;
import cn.iotab.iotkit.queue.rpc.factory.QueueRpcSetting;
import cn.iotab.iotkit.transport.api.common.message.JsonQueueMsg;
import cn.iotab.iotkit.transport.api.common.message.RpcRequestMessage;
import cn.iotab.iotkit.transport.api.common.message.RpcResponseMessage;
import cn.iotab.iotkit.transport.api.common.message.ToServerMessage;
import cn.iotab.iotkit.transport.api.common.message.ToTransportMessage;
import cn.iotab.iotkit.transport.api.common.message.codec.JsonMsgDecoder;
import lombok.extern.slf4j.Slf4j;

import javax.annotation.PreDestroy;

@Slf4j
public class KafkaTransportQueueFactory implements TransportQueueFactory {

    private final TransportQueueConfig queueConfig;

    private final KafkaAdmin kafkaAdmin;
    private final KafkaQueueFactory kafkaQueueFactory;
    private final DefaultQueueRpcFactory<JsonQueueMsg<RpcRequestMessage>, JsonQueueMsg<RpcResponseMessage>> defaultQueueRpcFactory;


    public KafkaTransportQueueFactory(TransportQueueConfig queueConfig) {
        this.queueConfig = queueConfig;
        KafkaSetting kafkaSetting = queueConfig.getKafkaSetting();

        kafkaAdmin = KafkaQueueAdminFactory.createQueueAdmin(kafkaSetting, queueConfig.getTopicSetting());
        kafkaQueueFactory = new KafkaQueueFactory(kafkaAdmin, kafkaSetting);
        defaultQueueRpcFactory = new DefaultQueueRpcFactory<>(
                kafkaAdmin,
                kafkaQueueFactory,
                new JsonMsgDecoder<>(RpcRequestMessage.class),
                new JsonMsgDecoder<>(RpcResponseMessage.class));
        //

    }

    @Override
    public QueueRequestTemplate<JsonQueueMsg<RpcRequestMessage>, JsonQueueMsg<RpcResponseMessage>> createTransportApiRequestTemplate(String rpcChannel) {
        QueueRpcSetting rpcSetting = queueConfig.getRpcSetting();
        rpcSetting.setRpcChannelId(rpcChannel);
        return defaultQueueRpcFactory.createRequestTemplate(rpcSetting);
    }

    @Override
    public QueueResponseTemplate<JsonQueueMsg<RpcRequestMessage>, JsonQueueMsg<RpcResponseMessage>> createTransportApiResponseTemplate(String rpcChannel) {
        QueueRpcSetting rpcSetting = queueConfig.getRpcSetting();
        rpcSetting.setRpcChannelId(rpcChannel);
        return defaultQueueRpcFactory.createResponseTemplate(rpcSetting);
    }

    @Override
    public QueueProducer<JsonQueueMsg<ToServerMessage>> createToServerMsgProducer(String topic) {
        QueueClientSetting clientSetting = getQueueClientSetting(topic, null);
        return kafkaQueueFactory.createProducer(clientSetting);
    }

    @Override
    public QueueProducer<JsonQueueMsg<ToTransportMessage>> createToTransportMsgProducer(String topic) {
        QueueClientSetting clientSetting = getQueueClientSetting(topic, null);
        return kafkaQueueFactory.createProducer(clientSetting);
    }

    private QueueClientSetting getQueueClientSetting(String topic, String group) {
        QueueClientSetting clientSetting = queueConfig.getClientSetting();
        clientSetting.setTopic(topic);
        clientSetting.setGroupId(group);
        return clientSetting;
    }

    @Override
    public QueueConsumer<JsonQueueMsg<ToTransportMessage>> createTransportNotificationsConsumer(String topic, String group) {
        QueueClientSetting clientSetting = getQueueClientSetting(topic, group);
        return kafkaQueueFactory.createConsumer(clientSetting, new JsonMsgDecoder<>(ToTransportMessage.class));
    }

    @Override
    public QueueConsumer<JsonQueueMsg<ToServerMessage>> createServerNotificationsConsumer(String topic, String group) {
        QueueClientSetting clientSetting = getQueueClientSetting(topic, group);
        return kafkaQueueFactory.createConsumer(clientSetting, new JsonMsgDecoder<>(ToServerMessage.class));
    }

    @PreDestroy
    public void destroy() {
        if (kafkaAdmin != null) {
            kafkaAdmin.destroy();
        }
    }
}